package com.we.risk.sa.stage3;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

public class SaEventLabelTest {
    public static final String RELEASEPROP =
            "risk/saevent/stagethird/saevent_stage3_kfk_cmp.properties";
    public static final String PROPSETS = "risk/saevent/stagethird/saevent_stage3_label.properties";
    public static final int LATESTDAYS = 30;
    public static final String KEYBY = "user_key";
    public static final String BEVTIME = "max_bev_time";
    public static final String RECVTIME = "max_receive_time";
    public static final String LABELK = "label_name";
    public static final String LABELV = "label_value";
    public static final String LABELT = "label_output_time";

    public static Logger LOG = LoggerFactory.getLogger(SaEventLabelTest.class);

    public static final String EVTAPEND = "$AppEnd";
    public static final String EVTAPSTT = "$AppStart";
    public static final String EVTPREPV = "$pageview";
    public static final String EVTAPVWSCRN = "AppViewScreen";
    public static final String EVTBANCLIK = "BannerClick";
    public static final String EVTCHOTCLK = "CashOutClick";
    public static final String EVTGETCODE = "GetCode";
    public static final String EVTHMSHWAPICARDCLK = "HomeShowApiCardClick";
    public static final String EVTHOMESHOW = "HomeShow";
    public static final String EVTLOANRSTAPCMP = "LoanResultApiComplete";
    public static final String EVTMDFLOANAMT = "ModifyLoanAmount";
    public static final String EVTMINESHOW = "MineShow";
    public static final String EVTMYFUNCETY = "MyFunctionEntry";
    public static final String EVTOPSVW = "OpenScreenView";
    public static final String EVTPV = "PageView";
    public static final String EVTSLCTINSTPL = "SelectInstallmentPlan";
    public static final String EVTSUPVIPPGV = "superVip_PageView";
    public static final String EVTSVPOPV = "ServicePopupView";
    public static final String EVTWDVW = "Windows_View";

    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final HashMap<String, String> transMap = initTransMap();

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            ArrayList<String> topicLists = new ArrayList<>();
            topicLists.add(srcKafkaTopic);
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("bootstrap.servers", srcKafkaBootStrapServer);
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setTopics(topicLists)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            DataStreamSource<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "Sa_event");

            SingleOutputStreamOperator<String> out =
                    input.uid("SaEvent-source")
                            .filter(new SaEventFilterFunc())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            String user_key = inputJson.getString("distinct_id");
                                            return user_key;
                                        }
                                    })
                            .flatMap(new SaEventFlatMapFunc());

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(SaEventLabelTest.class.toString());

        } catch (Exception e) {
            //            System.out.println(e.getMessage());
            LOG.error(e.getMessage());
            e.printStackTrace();
        }
    }

    private static class SaEventFilterFunc implements FilterFunction<String> {
        @Override
        public boolean filter(String value) throws Exception {
            //            boolean res = (value.contains(EVTMDFLOANAMT)  ||
            // value.contains(EVTSLCTINSTPL) || value.contains(EVTHMSHWAPICARDCLK));
            //            res = (res || value.contains(EVTLOANRSTAPCMP) ||
            // value.contains(EVTSUPVIPPGV) || value.contains(EVTAPSTT));
            //            res = (res || value.contains(EVTAPEND) || value.contains(EVTPV) ||
            // value.contains(EVTHOMESHOW));
            //            res = (res || value.contains(EVTMYFUNCETY) || value.contains(EVTBANCLIK)
            // || value.contains(EVTSVPOPV));
            //            res = (res || value.contains(EVTWDVW) || value.contains(EVTMINESHOW) ||
            // value.contains(EVTCHOTCLK));
            //            res = (res || value.contains(EVTAPVWSCRN) || value.contains(EVTOPSVW) ||
            // value.contains(EVTGETCODE));

            JSONObject inputJson = JSONObject.parseObject(value);
            Long recv_time = inputJson.getLong("recv_time");
            Long bev_time = inputJson.getLong("time");
            String event = inputJson.getString("event");
            String user_key = inputJson.getString("distinct_id");
            String project = inputJson.getString("project");

            //            System.out.println("project: " + (project == null ? "null" : project));

            if (recv_time == null
                    || recv_time.equals("")
                    || bev_time == null
                    || getStartTimeOfDay(recv_time) != getStartTimeOfDay(bev_time)
                    || event == null
                    || !transMap.containsKey(event)
                    || user_key == null
                    || project == null
                    || !project.equals("production")) {
                //                System.out.println("filter res false");
                return false;
            }
            //            System.out.println("filter res true");
            return true;
        }
    }

    private static class SaEventFlatMapFunc extends RichFlatMapFunction<String, String> {
        SimpleDateFormat sdf;
        /** 只存近LATESTDAYS内的数据 */
        MapState<Long, String> resMState;
        //        HashMap<String, String> transMap;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            sdf = new SimpleDateFormat(DATEFORMAT);
            resMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Long, String>(
                                            "resMState", Long.class, String.class));
            //            transMap = initTransMap();
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                JSONObject inputJson = JSONObject.parseObject(value);

                Long recvTime = inputJson.getLong("recv_time");
                String userKey = inputJson.getString("distinct_id");
                /** 脏数据过滤 SA 三期取消关于app_version限制 */
                if (userKey != null) {
                    String event = inputJson.getString("event");
                    Long bevTime = inputJson.getLong("time");

                    long mapKey = getStartTimeOfDay(bevTime);
                    long earliestTime = timeBeforeDays(recvTime, LATESTDAYS);
                    /** keep LATESTDAYS data in state */
                    cleanState(resMState, earliestTime);

                    String res = resMState.get(mapKey);

                    JSONObject stateJson = null;
                    if (res != null) {
                        stateJson = JSONObject.parseObject(res);
                    } else {
                        stateJson = initJsonObjState(PROPSETS);
                    }
                    SaEventOjbect saEventOjbect = new SaEventOjbect(event, bevTime, recvTime);
                    JSONObject resJson = (JSONObject) stateJson.clone();
                    //                    System.out.println("event: " + event);

                    labelProcess(transMap, saEventOjbect, resJson);

                    long max_bev_time = resJson.getLongValue(BEVTIME);
                    long max_receive_time = resJson.getLongValue(RECVTIME);
                    max_bev_time = (max_bev_time < bevTime ? bevTime : max_bev_time);
                    max_receive_time = (max_receive_time < recvTime ? recvTime : max_receive_time);
                    long label_output_time = System.currentTimeMillis();

                    resJson.put(KEYBY, userKey);
                    resJson.put(BEVTIME, max_bev_time);
                    resJson.put(RECVTIME, max_receive_time);
                    resJson.put(LABELT, label_output_time);

                    JSONObject outJson = new JSONObject();
                    outJson.put(KEYBY, userKey);
                    outJson.put(BEVTIME, max_bev_time);
                    outJson.put(RECVTIME, max_receive_time);
                    outJson.put(LABELT, label_output_time);
                    HashMap<String, Object> tmpLabel = new HashMap<>();

                    Iterator<String> iterator = resJson.keySet().iterator();
                    while (iterator.hasNext()) {
                        String key = iterator.next();
                        Object res_vae = resJson.get(key);
                        Object state_vae = stateJson.get(key);
                        if (!key.equals(KEYBY)
                                && !key.equals(BEVTIME)
                                && !key.equals(RECVTIME)
                                && !key.equals(LABELT)) {
                            if (state_vae != res_vae) {
                                tmpLabel.put(key, res_vae);
                            }
                        }
                    }
                    if (tmpLabel.size() > 0) {
                        Iterator<String> keyIterator = tmpLabel.keySet().iterator();
                        while (keyIterator.hasNext()) {
                            String next = keyIterator.next();
                            //                            System.out.println("##diff key: " + next);
                            outJson.put(LABELV, tmpLabel.get(next));
                            outJson.put(LABELK, next);
                            //                            System.out.println("##outJson user_key: "
                            // + userKey + " outJson size: " + outJson.size() + " outJsonData: " +
                            // outJson.toString());
                            out.collect(outJson.toString(SerializerFeature.WriteMapNullValue));
                        }
                    } else {
                        //                        System.out.println("##outJson user_key: " +
                        // userKey + " outJson size: " + outJson.size() + " outJsonData: " +
                        // outJson.toString());
                        out.collect(outJson.toString(SerializerFeature.WriteMapNullValue));
                    }

                    //                    System.out.println("##user_key: " + userKey + " ResJson
                    // size: " + resJson.size() + "JsonData: " + resJson.toString());
                    resMState.put(mapKey, resJson.toString(SerializerFeature.WriteMapNullValue));
                }
            } catch (Exception e) {
                LOG.error("SaEventFlatMapFunc excep: " + e.toString());
                LOG.error("SaEventFlatMapFunc excep input: " + value);
            }
        }

        private void labelProcess(
                HashMap<String, String> map, SaEventOjbect saEvent, JSONObject resJson) {
            String mapValue = map.get(saEvent.getEvent());
            if (mapValue != null) {
                long cntValue = resJson.getLongValue(mapValue + "_cnt");
                resJson.put(mapValue + "_cnt", cntValue + 1);

                long state_bev_time = resJson.getLongValue(mapValue + "_time");
                Long bev_time = saEvent.getBev_time();
                if (bev_time > state_bev_time) {
                    resJson.put(mapValue + "_time", bev_time);
                }
            }
        }

        private void cleanState(MapState<Long, String> state, long earliestTime) throws Exception {
            if (!state.isEmpty()) {
                ArrayList<Long> timeLists = new ArrayList<>();
                Iterator<Long> iterator = state.keys().iterator();
                while (iterator.hasNext()) {
                    Long timeKey = iterator.next();
                    if (timeKey < earliestTime) {
                        timeLists.add(timeKey);
                    }
                }
                int listSize = timeLists.size();
                for (int i = 0; i < listSize; i++) {
                    state.remove(timeLists.get(i));
                }
            }
        }

        private static JSONObject initJsonObjState(String propFile) throws IOException {
            JSONObject res = new JSONObject();
            InputStream resourceAsStream =
                    SaEventLabelTest.class.getClassLoader().getResourceAsStream(propFile);
            ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
            Set<Object> propSets = reader.getProperties().keySet();

            for (Object key : propSets) {
                String typeValue = reader.get((String) key);
                String[] splits = typeValue.split(":");
                if (splits.length == 2) {
                    String type = splits[0];
                    String value = splits[1];
                    jsonObjectSetValue(res, type, (String) key, value);
                } else {
                    LOG.error("split Error: " + splits.length);
                }
            }
            return res;
        }

        private static void jsonObjectSetValue(
                JSONObject jsonObj, String type, String key, String value) {
            if (!value.equals("null")) {
                switch (type) {
                    case "Byte":
                        jsonObj.put(key, Byte.parseByte(value));
                        break;
                    case "Long":
                        jsonObj.put(key, Long.parseLong(value));
                        break;
                    case "String":
                        jsonObj.put(key, value);
                        break;
                    case "Float":
                        jsonObj.put(key, Float.parseFloat(value));
                        break;
                    case "Integer":
                        jsonObj.put(key, Integer.parseInt(value));
                        break;
                    default:
                        LOG.error("Not support this type: " + type);
                        break;
                }
            } else {
                jsonObj.put(key, null);
            }
        }
    }

    private static HashMap<String, String> initTransMap() {
        HashMap<String, String> res = new HashMap<>();
        res.put(EVTAPEND, "md30_preset__append");
        res.put(EVTAPSTT, "md30_preset__appstart");
        res.put(EVTPREPV, "md30_preset__pageview");
        res.put(EVTAPVWSCRN, "md30_operation_appviewscreen");
        res.put(EVTBANCLIK, "md30_operation_bannerclick");
        res.put(EVTCHOTCLK, "md30_loan_cashoutclick");
        res.put(EVTGETCODE, "md30_client_getcode");
        res.put(EVTHMSHWAPICARDCLK, "md30_loan_supermarket_homeshowapicardclick");
        res.put(EVTHOMESHOW, "md30_client_homeshow");
        res.put(EVTLOANRSTAPCMP, "md30_loan_supermarket_loanresultapicomplete");
        res.put(EVTMDFLOANAMT, "md30_loan_modifyloanamount");
        res.put(EVTMINESHOW, "md30_client_mineshow");
        res.put(EVTMYFUNCETY, "md30_client_myfunctionentry");
        res.put(EVTOPSVW, "md30_operation_openscreenview");
        res.put(EVTPV, "md30_operation_pageview");
        res.put(EVTSLCTINSTPL, "md30_loan_selectinstallmentplan");
        res.put(EVTSUPVIPPGV, "md30_member_supervip_pageview");
        res.put(EVTSVPOPV, "md30_operation_servicepopupview");
        res.put(EVTWDVW, "md30_operation_windows_view");
        return res;
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(KEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }

    private static long timeBeforeDays(long now, int days) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(now);
        calendar.add(Calendar.DATE, -days);
        long res = calendar.getTimeInMillis();
        return res;
    }

    private static long getStartTimeOfDay(long now) {
        SimpleDateFormat sdf = new SimpleDateFormat(DATEFORMAT);
        String date = sdf.format(new Date(now));
        long res = 0;
        try {
            res = sdf.parse(date).getTime();
        } catch (Exception e) {
            LOG.error("sdf parse date error! time: " + now);
            e.printStackTrace();
        }
        return res;
    }
}
